package com.lzj.pagerank;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.KeyValueTextInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class PageRankJob {

    //收敛的指标
    private static double convergence = 0.0001;

    public static void main(String[] args) {
        //获取配置文件
        Configuration configuration = new Configuration(true);
        //跨平台执行
        configuration.set("mapreduce.app-submission.corss-paltform", "true");
        configuration.set("mapreduce.framework.name", "local");

        //计算迭代收敛的次数
        int runCount = 0;
        //开始执行迭代收敛
        while (true) {
            //计数累加
            runCount++;
            try {
                //向配置文件中设置一个变量
                configuration.setInt("runCount", runCount);
                //获取分布式文件系统
                FileSystem fs = FileSystem.get(configuration);
                //创建JOB,并设置基本信息
                Job job = Job.getInstance(configuration);
                job.setJarByClass(PageRankJob.class);
                job.setJobName("pagerank-" + runCount);
                //设置Mapper的输出类型
                job.setMapOutputKeyClass(Text.class);
                job.setMapOutputValueClass(Text.class);
                //设置Mapper和Reducer
                job.setMapperClass(PageRankMapper.class);
                job.setReducerClass(PageRankReducer.class);
                //使用了新的输入格式化类
                job.setInputFormatClass(KeyValueTextInputFormat.class);
                //设置读取数据的路径(第一次)
                Path inputPath = new Path("/lzj/pagerank/input/");
                //第二次之后读取数据,就是前一次的结果
                if (runCount > 1) {
                    inputPath = new Path("/lzj/pagerank/output/pr" + (runCount - 1));
                }
                //读取数据的路径
                FileInputFormat.addInputPath(job, inputPath);
                //本次输出路径
                Path outpath = new Path("/lzj/pagerank/output/pr" + runCount);
                if (fs.exists(outpath)) {
                    fs.delete(outpath, true);
                }
                //设置输出路径
                FileOutputFormat.setOutputPath(job, outpath);
                //提交任务,并获取本次任务是否成功
                boolean flag = job.waitForCompletion(true);
                if (flag) {
                    System.out.println("--------------------------success." + runCount);
                    //收敛值的和
                    long sum = job.getCounters().findCounter(Mycounter.CONVERGENCESUM).getValue();
                    double avgConvergence = sum / 4000.0;
                    //如果平均值达到收敛指标，停止循环
                    if (avgConvergence < convergence) {
                        break;
                    }
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

}

